// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <gtest/gtest.h>
#include <sstream>
#include <pollux/serializers/presto_serializer.h>

using namespace kumo::pollux;

class PrestoOutputStreamListenerTest : public ::testing::Test {
};

TEST_F(PrestoOutputStreamListenerTest, basic) {
    std::stringstream out;
    serializer::presto::PrestoOutputStreamListener streamlListener;
    auto os = std::make_unique<OStreamOutputStream>(&out, &streamlListener);

    std::string str1 = "this";
    std::string str2 = "is";
    std::string str3 = "hello world!";
    int length = str1.size() + str2.size() + str3.size();
    char codec = 4;
    int skipValue1 = 480;
    int skipValue2 = 8999;

    auto listener = dynamic_cast<serializer::presto::PrestoOutputStreamListener *>(
        os->listener());
    EXPECT_TRUE(listener != nullptr);

    bits::Crc32 crc;
    crc.process_bytes(&codec, sizeof(codec));
    crc.process_bytes(&length, sizeof(length));
    crc.process_bytes(str1.data(), str1.size());
    crc.process_bytes(str2.data(), str2.size());
    crc.process_bytes(str3.data(), str3.size());

    // We run following tests twice to test listener's reset capability.
    for (int i = 0; i < 2; ++i) {
        listener->reset();
        listener->resume();
        os->write(&codec, sizeof(codec));

        listener->pause();
        // Following two writes won't be reflected towards CRC computation.
        os->write(reinterpret_cast<char *>(&skipValue1), sizeof(skipValue1));
        os->write(reinterpret_cast<char *>(&skipValue2), sizeof(skipValue2));

        listener->resume();
        os->write(reinterpret_cast<char *>(&length), sizeof(length));
        os->write(str1.data(), str1.size());
        os->write(str2.data(), str2.size());
        os->write(str3.data(), str3.size());

        listener->pause();
        // Following two writes won't be reflected towards CRC computation.
        os->write(reinterpret_cast<char *>(&skipValue1), sizeof(skipValue1));
        os->write(reinterpret_cast<char *>(&skipValue2), sizeof(skipValue2));

        EXPECT_EQ(crc.checksum(), listener->crc().checksum());
    }
}
